package main

import (
	"container/list"
	"database/sql"
	"ever-checksum/ascii"
	"ever-checksum/config"
	"ever-checksum/errs"
	"ever-checksum/mysql"
	"ever-checksum/sqltext"
	"flag"
	"fmt"
	_ "github.com/go-sql-driver/mysql"
	_ "gopkg.in/eapache/queue.v1"
	"log/slog"
	"math/big"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"
)

const (
	LevelDebug = -4
	LevelInfo  = 0
	LevelWarn  = 4
	LevelError = 8
)

var (
	LoggerFile *slog.Logger // 文件日志
	LoggerStd  *slog.Logger // 命令行日志

	//wg sync.WaitGroup
	mu sync.Mutex

	srcDB *sql.DB // 源端连接
	tarDB *sql.DB // 目标端连接

	h bool
	v bool
	c string

	exitCh = make(chan bool) // 判断退出时机
	loadCh = make(chan bool) // 通知校验协程连接池情况
	loadOK bool

	tbIndex int
)

func init() {
	//参数初始化
	flag.BoolVar(&h, "h", false, "show this help")
	flag.BoolVar(&v, "v", false, "show version and exit")
	flag.StringVar(&c, "c", "", "config file (and path)")

	flag.Usage = usage
}

func usage() {
	fmt.Fprintf(os.Stderr, `my-checksum version: my-checksum-1.0.0
Usage: my-checksum [-hvc]

Options:
`)
	flag.PrintDefaults()
}

func initConn(srcdburl string, tardburl string, maxconn int, maxidleconn int) {
	if srcdburl == "" {
		fmt.Println("The SourceUrl is not configured, Please check the config file...")
		os.Exit(1)
	}
	if srcdburl == "" {
		fmt.Println("The TargetUrl is not configured, Please check the config file...")
		os.Exit(1)
	}
	if maxconn == 0 {
		maxconn = 100
	}
	if maxidleconn == 0 {
		maxidleconn = 10
	}

	srcDB, _ = sql.Open("mysql", srcdburl) //连接池初始化
	tarDB, _ = sql.Open("mysql", tardburl) //连接池初始化
	srcDB.SetMaxOpenConns(maxconn)
	tarDB.SetMaxOpenConns(maxconn)
	srcDB.SetMaxIdleConns(maxidleconn)
	tarDB.SetMaxIdleConns(maxidleconn)
	srcDB.SetConnMaxLifetime(time.Minute * 60)
	tarDB.SetConnMaxLifetime(time.Minute * 60)
	err1 := srcDB.Ping()
	err2 := tarDB.Ping()
	errs.CheckErr(err1)
	errs.CheckErr(err2)
}

func main() {
	flag.Parse()
	if h {
		flag.Usage()
		os.Exit(0)
	}
	if v {
		fmt.Println("my-checksum-1.0.0")
		os.Exit(0)
	}
	args := os.Args // 参数获取
	argNum := len(args)
	if argNum > 3 {
		fmt.Println("Wrong command line argument! Please check!")
	}

	var cof config.Conf
	dir, err := filepath.Abs(filepath.Dir(os.Args[0]))
	errs.CheckErr(err)
	if len(args) > 2 && args[1] == "-c" && argNum == 3 {
		cof = config.ReadConf(dir, args[2], "json")
	}

	logFile, err := os.OpenFile(cof.LogDir, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0666) //初始化日志
	if err != nil {
		fmt.Println("Log file failed to open: ", err)
		os.Exit(1)
	}
	defer func(logFile *os.File) {
		err = logFile.Close()
		if err != nil {
			fmt.Println(err.Error())
		}
	}(logFile)

	var lv = new(slog.LevelVar) // 根据参数设置日志级别
	switch cof.LogLevel {
	case LevelError:
		lv.Set(slog.LevelError)
	case LevelWarn:
		lv.Set(slog.LevelWarn)
	case LevelInfo:
		lv.Set(slog.LevelInfo)
	case LevelDebug:
		lv.Set(slog.LevelDebug)
	}
	lfh := slog.NewTextHandler(logFile, &slog.HandlerOptions{Level: lv})
	osh := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: lv})

	LoggerFile = slog.New(lfh) // 日志输出
	LoggerStd = slog.New(osh)  //命令行输出

	slog.SetDefault(LoggerFile)
	slog.SetDefault(LoggerStd)

	initConn(cof.SourceUrl, cof.TargetUrl, cof.MaxConnPool, cof.MaxConnIdlePool) // 初始化源端、目标端数据库连接
	LoggerFile.Info("The connection pool init is complete " + " MaxConnPool: " + strconv.Itoa(cof.MaxConnPool) + " MaxConnIdlePool: " + strconv.Itoa(cof.MaxConnIdlePool))

	tbIndex = 0

	wg := sync.WaitGroup{}
	maxGO := cof.ParallelT // 表级别并行，并行数由参数控制
	if cof.ParallelT > len(cof.TableArray) {
		maxGO = len(cof.TableArray)
	}
	wg.Add(maxGO)
	for i := 0; i < maxGO; i++ {
		go doCheckPre(cof, &wg)
	}
	wg.Wait()

	LoggerFile.Info("Congratulations! All the tables in config file have been checked!")
	os.Exit(0)
}

func doCheckPre(cof config.Conf, wg *sync.WaitGroup) {
	schema := cof.Schema
	ta := cof.TableArray
	var seq int
	defer wg.Done()
tableLoop:
	for {
		var tb config.Table
		mu.Lock()
		if tbIndex >= len(ta) {
			mu.Unlock()
			break tableLoop
		}
		tb = ta[tbIndex] // 从校验表数组中取一张表，后续生成校验任务
		seq = tbIndex    // 赋值给本协程内变量，tbIndex是全局变量
		tbIndex++        // 移动数组下标
		mu.Unlock()

		if cof.Checksum == "struct" {
			ok, x, y := compareTableStruct(tb.Tablename, schema)
			LoggerFile.Info("Struct check: " + tb.Tablename + " , Consistent?: " + strconv.FormatBool(ok) + " ")
			fmt.Println("Struct check: " + tb.Tablename + " , Consistent?: " + strconv.FormatBool(ok) + " ")
			if !ok {
				LoggerFile.Info("Struct check: " + tb.Tablename + " , Struct: " + x)
				LoggerFile.Info("Struct check: " + tb.Tablename + " , Struct: " + y)
				fmt.Println("Struct check: " + tb.Tablename + " , Struct: " + x)
				fmt.Println("Struct check: " + tb.Tablename + " , Struct: " + y)
			}
			continue tableLoop
		} else if cof.Checksum == "data" {
			var csTab CSTab // // 配置参数读取判断，避免每次传递 config.Conf
			if tb.PalallelS != 0 {
				csTab.PalallelS = tb.PalallelS
			} else {
				csTab.PalallelS = cof.PalallelS
			}

			if tb.ChunkSize != 0 {
				csTab.ChunkSize = tb.ChunkSize
			} else {
				csTab.ChunkSize = cof.ChunkSize
			}

			csTab.DelaySecond = cof.Delay
			csTab.Schema = schema
			csTab.Seq = strconv.Itoa(seq + 1)
			csTab.Interval = cof.Interval

			csTab.GridFlag = cof.GridFlag
			if cof.SDataServer == "" || cof.TDataServer == "" {
				fmt.Println("sdataserver & tdataserver not found in conf file...")
				os.Exit(1)
			}
			csTab.SDataServer = cof.SDataServer
			csTab.TDataServer = cof.TDataServer
			csTab.Database = cof.Database // 未设置database时，与schema值相同
			if csTab.Database == "" {
				csTab.Database = csTab.Schema
			}
			if tb.Tablename == "" {
				LoggerFile.Error("Task: " + csTab.Seq + ", Table name is empty, Please check the config file...")
				fmt.Println("Task: " + csTab.Seq + ", Table name is empty, Please check the config file...")
				continue tableLoop
			}
			csTab.Table = tb.Tablename

			if tb.SplitColumn == "" { // 未指定索引&切分列，则查询表索引情况选择合适的索引
				csTab.OrderBy = queryForIndexCols(schema, tb.Tablename, "PRIMARY") // 查询表的主键情况，默认从源端操作
				if csTab.OrderBy == nil {
					fmt.Println("Task: " + csTab.Seq + ", Table: " + schema + "." + tb.Tablename + "'s find no primary key...")
					LoggerFile.Error("Task: " + csTab.Seq + ", Table: " + schema + "." + tb.Tablename + "'s find no primary key...")
					break tableLoop
				}
				csTab.Column = csTab.OrderBy[0] // 使用主键索引第一列切分
			} else {
				if tb.SplitColumn != "" {
					fmt.Println("Task: " + csTab.Seq + ", Table: " + schema + "." + tb.Tablename + "'s split column are configured, check will be based on split column...")
					// 判断指定的拆分列是不是主键中的列
					sqlColIsEx := fmt.Sprintf("SHOW INDEX FROM %s.%s WHERE Column_name='%s' AND Key_name='PRIMARY'", schema, tb.Tablename, tb.SplitColumn)
					if len(mysql.QueryForSet(srcDB, sqlColIsEx)) == 0 {
						LoggerFile.Error("Task: " + csTab.Seq + ", Table: " + schema + "." + tb.Tablename + " configured split column: " + tb.SplitColumn + " is not a primary key column...")
						fmt.Println("Task: " + csTab.Seq + ", Table: " + schema + "." + tb.Tablename + " configured split column: " + tb.SplitColumn + " is not a primary key column...")
						continue tableLoop
					}
					csTab.Column = tb.SplitColumn
					csTab.OrderBy = queryForSplitIndex(schema, tb.Tablename, csTab.Column)
					if csTab.OrderBy == nil {
						fmt.Println("Task: " + csTab.Seq + ", Table: " + schema + "." + tb.Tablename + "'s find no index on " + csTab.Column + "...")
						LoggerFile.Error("Task: " + csTab.Seq + ", Table: " + schema + "." + tb.Tablename + "'s find no index on " + csTab.Column + "...")
						continue tableLoop
					}
				}
			}
			csTab.Columns, csTab.ColumnTypes, _ = queryForColumnsAndType(schema, tb.Tablename, csTab.OrderBy, "")
			for _, v := range csTab.ColumnTypes {
				csTab.ColumnQuots = append(csTab.ColumnQuots, quotM(v))
			}
			// 指定了额外列与值
			if tb.ValueColumnK != "" && tb.ValueColumnV != "" { // 使用指定索引，使用指定列
				sqlColIsEx := fmt.Sprintf("%s count(1) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA='%s' AND TABLE_NAME='%s' AND COLUMN_NAME='%s'", "SELECT ", schema, tb.Tablename, tb.ValueColumnK)
				if len(mysql.QueryForSet(srcDB, sqlColIsEx)) == 0 {
					LoggerFile.Error("Task: " + csTab.Seq + ", Table: " + schema + "." + tb.Tablename + "configured value column(valuecolumnk): " + tb.ValueColumnK + " is not exist...")
					fmt.Println("Task: " + csTab.Seq + ", Table: " + schema + "." + tb.Tablename + "configured value column(valuecolumnk): " + tb.ValueColumnK + " is not exist...")
					//break
					continue tableLoop
				}
				csTab.ValueColumnK = tb.ValueColumnK
				csTab.ValueColumnV = tb.ValueColumnV
				_, _, csTab.ValueColumnType = queryForColumnsAndType(schema, tb.Tablename, []string{}, csTab.ValueColumnK)
			}
			if tb.Partition != "" {
				csTab.Table += " PARTITION(" + tb.Partition + ")"
			}
			if csTab.Column != "" && csTab.Columns != "" && len(csTab.ColumnTypes) != 0 {
				genCheckSumJob(csTab)
			} else {
				LoggerFile.Error("Task: " + csTab.Seq + ", Table: " + schema + "." + tb.Tablename + " failed to get (index name)/(split column), please check the table...")
				fmt.Println("Task: " + csTab.Seq + ", Table: " + schema + "." + tb.Tablename + " failed to get (index name)/(split column), please check the table...")
				continue tableLoop
			}
		}
	}
}

func genCheckSumJob(csTab CSTab) {
	var allQuerySql [][]string
	var rangeList list.List
	cntCh := make(chan CountTuple, 1) // 校验数量统计channel，校验协程将校验通过的数据量写入该channel
	wc := sync.WaitGroup{}

	var colSubQuery string // 指定了条件，例如分区表指定日期时间
	if csTab.ValueColumnK != "" && csTab.ValueColumnV != "" {
		if mysql.IsInt64Type(csTab.ValueColumnType) || mysql.IsBigIntType(csTab.ValueColumnType) {
			colSubQuery = fmt.Sprintf(" AND %s=%s ", csTab.ValueColumnK, csTab.ValueColumnV)
		} else if mysql.IsStringType(csTab.ValueColumnType) {
			colSubQuery = fmt.Sprintf(" AND %s=%s%s%s ", csTab.ValueColumnK, "'", csTab.ValueColumnV, "'")
		}
	}

	var iAddr int // 确定指定列在索引中的位置
	var pAddr int // 确定指定值的列在索引中的位置
	for i, v := range csTab.OrderBy {
		if csTab.Column == v {
			iAddr = i
		}
		if csTab.ValueColumnK == v {
			pAddr = i
		}
	}
	// 两种情况走这个if-condition
	// 第一种：索引列只有一列，且为数值类型，可以切分并行
	// 第二种：拆分键被指定，且拆分键类型是INT，且拆分键是第二列，且制定了第一列值列或指定的拆分键为非数值类型
	if (len(csTab.OrderBy) == 1 && csTab.ColumnTypes[0] == "INT") || (csTab.ColumnTypes[iAddr] == "INT" && csTab.ValueColumnK != "" && csTab.ValueColumnV != "" && pAddr == 0 && iAddr == 1) {
		x, y := getRange(csTab, colSubQuery) // 查询索引指定列最小值、最大值
		if x != nil && y != nil && x != "" && y != "" {
			if mysql.IsInt64Type(csTab.ColumnTypes[0]) { // 根据int类型索引进行任务切割，获取切分条件(单列索引)
				xint64, xerr := strconv.ParseInt(x.(string), 10, 64)
				yint64, yerr := strconv.ParseInt(y.(string), 10, 64)
				errs.CheckErr(xerr)
				errs.CheckErr(yerr)
				rangeList = ascii.WrapRangeInt64Arr(ascii.DoLongSplit(xint64, yint64, csTab.PalallelS), csTab.Column)
			}
			if mysql.IsBigIntType(csTab.ColumnTypes[0]) { // 根据bigint类型索引进行任务切割，获取切分条件(单列索引)
				xint64, xerr := strconv.ParseInt(x.(string), 10, 64)
				yint64, yerr := strconv.ParseInt(y.(string), 10, 64)
				errs.CheckErr(xerr)
				errs.CheckErr(yerr)
				rangeList = ascii.WrapRangeBigIntArr(ascii.DoBigIntSplit(big.NewInt(xint64), big.NewInt(yint64), csTab.PalallelS), csTab.Column)
			}
			if rangeList.Len() != 0 { // 根据切分情况生成crc32语句
				for e := rangeList.Front(); e != nil; e = e.Next() {
					var subQueryList []string
					subQueryList = append(subQueryList, "split")
					subQueryList = append(subQueryList, sqltext.BuildSubQuerySql(strings.Join(csTab.OrderBy, ","), csTab.Columns, csTab.Table, "", csTab.GridFlag, csTab.Database)+" where 1=1")
					subQueryList = append(subQueryList, e.Value.(string))
					subQueryList = append(subQueryList, fmt.Sprintf(" %s order by %s limit %d", colSubQuery, strings.Join(csTab.OrderBy, ","), csTab.ChunkSize))
					allQuerySql = append(allQuerySql, subQueryList)
				}
			}
		} else {
			LoggerFile.Info("Task: " + csTab.Seq + ", Table: " + csTab.Database + "." + csTab.Table + " min/max value is null...")
			LoggerStd.Info("Task: " + csTab.Seq + ", Table: " + csTab.Database + "." + csTab.Table + " min/max value is null...")

			// 确认两端行数是否一致
			countSQL := fmt.Sprintf("%s %s %s %s.%s", "SELECT ", " count(1) as cnt ", " FROM ", csTab.Database, csTab.Table)
			var sqlCountTar string
			if csTab.GridFlag {
				sqlCountTar = "dbscale execute on dataserver " + csTab.TDataServer + " \"" + countSQL + "\""
			} else {
				sqlCountTar = countSQL
			}
			LoggerFile.Info("Task: " + csTab.Seq + ", Count SQL: " + sqlCountTar)

			var resultS []map[string]string
			var resultT []map[string]string
			wgCount := sync.WaitGroup{}
			wgCount.Add(1)
			go func() {
				defer wgCount.Done()
				resultS = mysql.QueryForSet(srcDB, sqlCountTar) // 源端查询
			}()
			wgCount.Add(1)
			go func() {
				defer wgCount.Done()
				resultT = mysql.QueryForSet(tarDB, countSQL)
			}()
			wgCount.Wait()
			checkOk := false
			if resultS[0]["cnt"] == resultT[0]["cnt"] {
				checkOk = true
			}
			wc.Add(1)
			go func() {
				defer wc.Done()
				cntCh <- CountTuple{
					cnt:  0,
					ok:   checkOk,
					exit: true,
				}
			}()
		}
	} else { // 不能切分并行的情况，查询每个chunk后，使用最后一行索引列数值确定下一个chunk
		var demoSQL string
		var sqlFirstSrc string
		var sqlFirstTar string
		demoSQL = fmt.Sprintf("%s %s, CRC32(concat_ws('#', %s)) as checksum from %s.%s where 1=1 ", "SELECT ", strings.Join(csTab.OrderBy, ", "), csTab.Columns, csTab.Database, csTab.Table)
		firstSQL := demoSQL + colSubQuery + " order by " + strings.Join(csTab.OrderBy, ", ") + " limit 1"
		if csTab.GridFlag {
			sqlFirstSrc = "dbscale execute on dataserver " + csTab.SDataServer + " \"" + firstSQL + "\""
			sqlFirstTar = "dbscale execute on dataserver " + csTab.TDataServer + " \"" + firstSQL + "\""
		} else {
			sqlFirstSrc = firstSQL
			sqlFirstTar = firstSQL
		}
		var colValues []string
		var checkOk bool
		exitFlag := false
		var fCheckCnt int
		var resultS []map[string]string
		var resultT []map[string]string

		wf := sync.WaitGroup{}
		wf.Add(1)
		go func() {
			defer wf.Done()
			resultS = mysql.QueryForSet(srcDB, sqlFirstSrc) // 源端查询
		}()
		wf.Add(1)
		go func() {
			defer wf.Done()
			resultT = mysql.QueryForSet(tarDB, sqlFirstTar)
		}()
		wf.Wait()
		if len(resultS) != 1 { // 查询结果集是0行,直接报告channel后退出
			fCheckCnt = 0
			if len(resultT) == 1 { // 不一致，写入channel
				checkOk = false
			} else { // 都是0行，数据一致写入channel
				checkOk = true
			}
			exitFlag = true
		} else { // 校验第一行数据
			fCheckCnt = 1
			if resultS[0]["checksum"] == resultT[0]["checksum"] {
				for _, v := range csTab.OrderBy {
					colValues = append(colValues, resultS[0][v])
				}
				checkOk = true
			} else {
				checkOk = false
				exitFlag = true
			}
		}
		wc.Add(1)
		go func() { // 将第一行的校验结果写入channel
			defer wc.Done()
			cntCh <- CountTuple{
				cnt:  fCheckCnt,
				ok:   checkOk,
				exit: exitFlag,
			}
		}()
		rangeList.PushBack(genCond(csTab.OrderBy, colValues, csTab.ColumnQuots)) // 一列到多列的索引列条件生成
		LoggerStd.Info("Task: " + csTab.Seq + ", Table: " + csTab.Database + "." + csTab.Table + " complete generate split SQLs...")
		// 根据切分情况生成crc32语句
		if rangeList.Len() != 0 {
			for e := rangeList.Front(); e != nil; e = e.Next() {
				if e.Value.(string) != "" {
					var subQueryList []string
					subQueryList = append(subQueryList, "paging")
					subQueryList = append(subQueryList, sqltext.BuildSubQuerySql(strings.Join(csTab.OrderBy, ","), csTab.Columns, csTab.Table, "", csTab.GridFlag, csTab.Database)+" where 1=1")
					subQueryList = append(subQueryList, e.Value.(string))
					subQueryList = append(subQueryList, fmt.Sprintf(" %s order by %s limit %d", colSubQuery, strings.Join(csTab.OrderBy, ","), csTab.ChunkSize))
					allQuerySql = append(allQuerySql, subQueryList)
				}
			}
		}
	}
	countCheck := 0
	LoggerStd.Info("Task: " + csTab.Seq + ", Table: " + csTab.Database + "." + csTab.Table + " start check data now...")
	jobCnt := len(allQuerySql)
	wc.Add(1) // 代表其中一个是数据统计协程
	checkStart := time.Now().Unix()
	i := 0 // 用于判断校验任务数量，全部校验结束后，即可关闭channel
	var pass bool
	go func() { // 异步读取校验进度用于展示
		defer wc.Done()
		interValStart := time.Now().Unix()
	countLoop:
		for {
			ct := <-cntCh
			countCheck += ct.cnt
			pass = ct.ok
			interValStop := time.Now().Unix() // 控制命令行输出频率
			if (interValStop - interValStart) > csTab.Interval {
				fmt.Println("Task: " + csTab.Seq + ", Table: " + csTab.Database + "." + csTab.Table + " check is running, rows checked: " + strconv.Itoa(countCheck))
				interValStart = interValStop
			}

			if ct.exit { // 判断退出时机
				i++
				if jobCnt == i || jobCnt == 0 {
					checkStop := time.Now().Unix()
					fmt.Println("Task: " + csTab.Seq + ", Table: " + csTab.Database + "." + csTab.Table + " check complete, check ok rows: " + strconv.Itoa(countCheck) + " check OK?: " + strconv.FormatBool(pass) + " Duration: " + strconv.FormatInt(checkStop-checkStart, 10) + " second")
					LoggerFile.Info("Task: " + csTab.Seq + ", Table: " + csTab.Database + "." + csTab.Table + " check complete, check ok rows: " + strconv.Itoa(countCheck) + " check OK?: " + strconv.FormatBool(pass) + " Duration: " + strconv.FormatInt(checkStop-checkStart, 10) + " second")
					close(cntCh) // 所有并行job退出后关闭channel  退出循环
					break countLoop
				}
			} else {
				continue countLoop
			}
		}
	}()

	fmt.Println("Task: " + csTab.Seq + ", Table: " + csTab.Database + "." + csTab.Table + " check begins, time: " + time.Now().Format("2006-01-02 15:04:05 MST Mon"))
	for j := 0; j < len(allQuerySql); j++ {
		fmt.Println("Task: " + csTab.Seq + ", Table: " + csTab.Database + "." + csTab.Table + " sub check task begins, times: " + time.Now().Format("2006-01-02 15:04:05 MST Mon"))
		wc.Add(1)
		go doRealCheckSum(allQuerySql[j], csTab, &wc, cntCh) // 表内校验任务并行
	}
	wc.Wait()
}

func doRealCheckSum(sqlArray []string, csTab CSTab, wc *sync.WaitGroup, cntCh chan CountTuple) {
	defer wc.Done()
	srcChecksumCh := make(chan CheckSumTuple) // 异步查询将结果集通过这个channel传给查询的主线程
	dstChecksumCh := make(chan CheckSumTuple)
	defer close(srcChecksumCh)
	defer close(dstChecksumCh)
	i := 0
	sqlCheck := ""

	var tmpValues []string // 循环分页查询，直至返回数据量小于chunk size
limitLoop:
	for {
		if len(sqlArray) == 4 {
			sqlCheck = sqlRebuild(sqlArray, tmpValues, csTab.OrderBy, csTab.ColumnQuots) // 语句中条件处理，条件根据上一次查询进行调整，用于优化limit分页性能
			sqlCheckSrc := ""
			sqlCheckTar := ""
			if csTab.GridFlag {
				sqlCheckSrc = "dbscale execute on dataserver " + csTab.SDataServer + " \"" + sqlCheck + "\""
				sqlCheckTar = "dbscale execute on dataserver " + csTab.TDataServer + " \"" + sqlCheck + "\""
			} else {
				sqlCheckSrc = sqlCheck
				sqlCheckTar = sqlCheck
			}
			var wg sync.WaitGroup
			wg.Add(1)
			go func() { // 查询源端，写入channel
				defer wg.Done()
				colValues, srcSet := mysql.QueryForCheckResult(csTab.OrderBy, srcDB, sqlCheckSrc)
				if len(colValues) > 0 {
					tmpValues = convertArray(colValues)
				}
				if len(srcSet) < csTab.ChunkSize {
					srcChecksumCh <- CheckSumTuple{flag: "break", checksum: srcSet}
				} else if len(srcSet) == csTab.ChunkSize {
					srcChecksumCh <- CheckSumTuple{flag: "limitEOF", checksum: srcSet}
				}
			}()

			wg.Add(1)
			go func() { // 查询目标端，写入channel
				defer wg.Done()
				_, dstSet := mysql.QueryForCheckResult(csTab.OrderBy, tarDB, sqlCheckTar)
				if len(dstSet) < csTab.ChunkSize {
					dstChecksumCh <- CheckSumTuple{flag: "break", checksum: dstSet}
				} else if len(dstSet) == csTab.ChunkSize {
					dstChecksumCh <- CheckSumTuple{flag: "limitEOF", checksum: dstSet}
				}
			}()
			srcCheckSum := <-srcChecksumCh
			dstCheckSum := <-dstChecksumCh
			wg.Wait()                          // 等待校验查询返回
			if len(srcCheckSum.checksum) > 0 { // 校验查询的checksum
				if CompareCheckSum(csTab.Database+"."+csTab.Table, srcCheckSum, dstCheckSum, cntCh) {
					// 返回数据量小于ChunkSize，及时退出
					if srcCheckSum.flag == "limitEOF" {
						i++
					} else if srcCheckSum.flag == "break" {
						break limitLoop
					}
					continue limitLoop
				} else {
					if csTab.DelaySecond > 0 { // 是否设置了容忍延迟时间
						start := time.Now().Unix()
						stop := start
					delayLoop:
						for { // 读秒循环查询目标端，源端不动，直至数据一致或者计时结束
							if int64(csTab.DelaySecond) >= (stop - start) {
								go func() { // 异步查询目标端，写入channel
									_, dstSet := mysql.QueryForCheckResult(csTab.OrderBy, tarDB, sqlCheckTar)
									if len(dstSet) < csTab.ChunkSize {
										dstChecksumCh <- CheckSumTuple{flag: "break", checksum: dstSet}
									} else if len(dstSet) == csTab.ChunkSize {
										dstChecksumCh <- CheckSumTuple{flag: "limitEOF", checksum: dstSet}
									}
								}()

								dstCheckSum = <-dstChecksumCh
								if CompareCheckSum(csTab.Database+"."+csTab.Table, srcCheckSum, dstCheckSum, cntCh) {
									if dstCheckSum.flag == "limitEOF" {
										i++
									} else if dstCheckSum.flag == "break" {
										break limitLoop
									}
									continue limitLoop // 校验通过且未结束，继续加一个分页循环
								} else {
									time.Sleep(100 * time.Millisecond) // 休眠100毫秒，避免快速循环打满CPU
									stop = time.Now().Unix()
									continue delayLoop // 校验失败，延迟时间内，继续循环校验
								}
							} else {
								break delayLoop
							}
						}
					}

					LoggerFile.Info("Task: " + csTab.Seq + ", There are inconsistencies in the check, SQL: " + sqlCheck)
					go func() {
						cntCh <- CountTuple{
							cnt:  0,
							ok:   false,
							exit: true, // 校验不一致退出
						}
					}()
					break limitLoop
				}
			} else {
				LoggerFile.Info("Task: " + csTab.Seq + ", SOURCE/TARGET DB query empty, SQL: " + sqlCheck)
				go func() {
					cntCh <- CountTuple{
						cnt:  0,
						ok:   len(srcCheckSum.checksum) == len(dstCheckSum.checksum),
						exit: true,
					}
				}()
				break limitLoop
			}
		} else {
			LoggerFile.Error("Task: " + csTab.Seq + ", Check SQL incorrect, please check! SQL: " + sqlCheck)
			go func() {
				cntCh <- CountTuple{
					cnt:  0,
					ok:   false,
					exit: true,
				}
			}()
			break limitLoop
		}
	}
}

func genCond(cNs []string, cVs []string, q []string) string {
	a := " AND "
	o := " OR "

	whereSQL := ""
	if len(cVs) == 0 { // 列值数组为空直接返回
		return ""
	}
	sql0 := cNs[0] + " = " + q[0] + cVs[0] + q[0]
	sqls := "(" + cNs[0] + " > " + q[0] + cVs[0] + q[0] + ")"
	switch len(cVs) {
	case 1: // 主键一列

	case 2: // 主键两列
		whereSQL += "(" + sql0 + a + cNs[1] + " > " + q[1] + cVs[1] + q[1] + ")" + o
	case 3: // 主键三列
		whereSQL += "(" + sql0 + a + cNs[1] + " = " + q[1] + cVs[1] + q[1] + a + cNs[2] + " > " + q[2] + cVs[2] + q[2] + ")" + o
		whereSQL += "(" + sql0 + a + cNs[1] + " > " + q[1] + cVs[1] + q[1] + ")" + o
	case 4: // 主键四列
		whereSQL += "(" + sql0 + a + cNs[1] + " = " + q[1] + cVs[1] + q[1] + a + cNs[2] + " = " + q[2] + cVs[2] + q[2] + a + cNs[3] + " > " + q[3] + cVs[3] + q[3] + ")" + o
		whereSQL += "(" + sql0 + a + cNs[1] + " = " + q[1] + cVs[1] + q[1] + a + cNs[2] + " > " + q[2] + cVs[2] + q[2] + ")" + o
		whereSQL += "(" + sql0 + a + cNs[1] + " > " + q[1] + cVs[1] + q[1] + ")" + o
	case 5: // 主键五列
		whereSQL += "(" + sql0 + a + cNs[1] + " = " + q[1] + cVs[1] + q[1] + a + cNs[2] + " = " + q[2] + cVs[2] + q[2] + a + cNs[3] + " = " + q[3] + cVs[3] + q[3] + a + cNs[4] + " > " + q[4] + cVs[4] + q[4] + ")" + o
		whereSQL += "(" + sql0 + a + cNs[1] + " = " + q[1] + cVs[1] + q[1] + a + cNs[2] + " = " + q[2] + cVs[2] + q[2] + a + cNs[3] + " > " + q[3] + cVs[3] + q[3] + ")" + o
		whereSQL += "(" + sql0 + a + cNs[1] + " = " + q[1] + cVs[1] + q[1] + a + cNs[2] + " > " + q[2] + cVs[2] + q[2] + ")" + o
		whereSQL += "(" + sql0 + a + cNs[1] + " > " + q[1] + cVs[1] + q[1] + ")" + o
	case 6: // 主键六列，用这么多列做主键，多少有点欠揍
		whereSQL += "(" + sql0 + a + cNs[1] + " = " + q[1] + cVs[1] + q[1] + a + cNs[2] + " = " + q[2] + cVs[2] + q[2] + a + cNs[3] + " = " + q[3] + cVs[3] + q[3] + a + cNs[4] + " = " + q[4] + cVs[4] + q[4] + a + cNs[5] + " > " + q[5] + cVs[5] + q[5] + ")" + o
		whereSQL += "(" + sql0 + a + cNs[1] + " = " + q[1] + cVs[1] + q[1] + a + cNs[2] + " = " + q[2] + cVs[2] + q[2] + a + cNs[3] + " = " + q[3] + cVs[3] + q[3] + a + cNs[4] + " > " + q[4] + cVs[4] + q[4] + ")" + o
		whereSQL += "(" + sql0 + a + cNs[1] + " = " + q[1] + cVs[1] + q[1] + a + cNs[2] + " = " + q[2] + cVs[2] + q[2] + a + cNs[3] + " > " + q[3] + cVs[3] + q[3] + ")" + o
		whereSQL += "(" + sql0 + a + cNs[1] + " = " + q[1] + cVs[1] + q[1] + a + cNs[2] + " > " + q[2] + cVs[2] + q[2] + ")" + o
		whereSQL += "(" + sql0 + a + cNs[1] + " > " + q[1] + cVs[1] + q[1] + ")" + o
	default:
	}
	whereSQL += sqls
	return whereSQL
}

func CompareCheckSum(schTab string, srcChecksum CheckSumTuple, dstChecksum CheckSumTuple, cntCh chan CountTuple) bool {
	if strings.Join(srcChecksum.checksum, "") == strings.Join(dstChecksum.checksum, "") { // 拼接字符串形式对比
		lChecksum := len(srcChecksum.checksum) // 使用对比切片方式对比
		go func() {
			exitJob := false
			if srcChecksum.flag == "break" {
				exitJob = true
			}
			cntCh <- CountTuple{
				cnt:  lChecksum,
				ok:   true,
				exit: exitJob,
			}
		}()
		LoggerFile.Debug(schTab + " is checking, batch rows: " + strconv.Itoa(lChecksum))
		return true
	} else {
		return false
	}
}

func queryForSplitIndex(schema string, table string, colName string) []string {
	var colNames []string
	restIndex := mysql.QueryForSet(srcDB, "SHOW INDEX FROM "+schema+"."+table+" WHERE Column_name='"+colName+"'") // 以源端索引为准
	if len(restIndex) > 0 {
		keyName := restIndex[0]["Key_name"] // 索引
		colNames = queryForIndexCols(schema, table, keyName)
	} else {
		return nil
	}
	return colNames
}

func queryForIndexCols(schema string, table string, keyName string) []string {
	var colName string // 索引列列名
	var colNames []string
	restIndex := mysql.QueryForSet(srcDB, "SHOW INDEX FROM "+schema+"."+table+" WHERE Key_name='"+keyName+"'") // 以源端索引为准
	for i := 0; i < len(restIndex); i++ {
		colName = restIndex[i]["Column_name"] // 索引列列名
		colNames = append(colNames, colName)
	}
	return colNames
}

func queryForColumnsAndType(schema string, table string, cols []string, col2 string) (string, []string, string) {
	sqlColType := "SELECT COLUMN_NAME,COLUMN_TYPE from information_schema.columns where TABLE_SCHEMA='" + schema + "' and TABLE_NAME='" + table + "'"
	var colsType []string
	var col2Type string
	var columns string
	var columnsTmp []string
	columnsSet := mysql.QueryForSet(srcDB, sqlColType) // 查询源端
	if len(columnsSet) == 0 {
		LoggerFile.Debug("Query for " + schema + "." + table + "'s column type, the result set is empty...")
		return "", nil, ""
	} else {
		for i := 0; i < len(columnsSet); i++ {
			columnsTmp = append(columnsTmp, "`"+columnsSet[i]["COLUMN_NAME"]+"`")
			// 多列类型放数组
			if len(cols) > 0 {
				for _, value := range cols {
					if value == columnsSet[i]["COLUMN_NAME"] {
						var temType string
						if strings.Contains(columnsSet[i]["COLUMN_TYPE"], "(") {
							temType = strings.ToUpper(columnsSet[i]["COLUMN_TYPE"][:strings.Index(columnsSet[i]["COLUMN_TYPE"], "(")])
						} else {
							temType = strings.ToUpper(columnsSet[i]["COLUMN_TYPE"])
						}
						colsType = append(colsType, temType)
					}
				}
			}
			if columnsSet[i]["COLUMN_NAME"] == col2 {
				if strings.Contains(columnsSet[i]["COLUMN_TYPE"], "(") {
					col2Type = strings.ToUpper(columnsSet[i]["COLUMN_TYPE"][:strings.Index(columnsSet[i]["COLUMN_TYPE"], "(")])
				} else {
					col2Type = strings.ToUpper(columnsSet[i]["COLUMN_TYPE"])
				}
			}
		}
		columns = strings.Join(columnsTmp, ",")
	}
	return columns, colsType, col2Type
}

func compareTableStruct(tabName string, schema string) (bool, string, string) {
	sqlTabExi := fmt.Sprintf("%s %s %s %s %s %s%s%s %s %s%s%s %s %s%s%s ", "SELECT", "TABLE_NAME", "FROM", "information_schema.tables", "WHERE", "TABLE_SCHEMA='", schema, "'", "AND", "TABLE_NAME='", tabName, "'", "AND", "TABLE_TYPE='", "BASE TABLE", "'")
	exiS := mysql.QueryForSet(srcDB, sqlTabExi)
	exiD := mysql.QueryForSet(tarDB, sqlTabExi)
	if exiS != nil && exiD != nil {
		sqlGetTableStruct := fmt.Sprintf("%s %s%s%s", "show create table", schema, ".", tabName)
		srcT := mysql.QueryForSet(srcDB, sqlGetTableStruct)
		dstT := mysql.QueryForSet(tarDB, sqlGetTableStruct)
		if len(srcT) == 1 && len(dstT) == 1 {
			x := srcT[0]["Create Table"]
			y := dstT[0]["Create Table"]
			if x == y {
				return true, "", ""
			} else {
				return false, x, y
			}
		}
	} else {
		if exiS == nil {
			LoggerFile.Error("Table " + schema + "." + tabName + " doesn't exist in source DB...")
			fmt.Println("Table " + schema + "." + tabName + " doesn't exist in source DB...")
		}
		if exiD == nil {
			LoggerFile.Error("Table " + schema + "." + tabName + " doesn't exist in target DB...")
			fmt.Println("Table " + schema + "." + tabName + " doesn't exist in target DB...")
		}
		return false, "", ""
	}
	return false, "", ""
}

func getRange(csTab CSTab, colSubQuery string) (interface{}, interface{}) {
	var sqlRange string
	if csTab.GridFlag {
		// 连接具体DB节点的server，以源端为准
		//sqlCont := "select count(1) as cnt, min(" + csTab.Column + ") as min, max(" + csTab.Column + ") as max from " + csTab.Database + "." + csTab.Table
		sqlCont := "select min(" + csTab.Column + ") as min, max(" + csTab.Column + ") as max from " + csTab.Database + "." + csTab.Table
		if colSubQuery != "" {
			sqlCont += " WHERE 1=1 " + colSubQuery
		}
		sqlRange = "dbscale execute on dataserver " + csTab.SDataServer + " \"" + sqlCont + "\""
	} else {
		//sqlCont := "select count(1) as cnt, min(" + csTab.Column + ") as min, max(" + csTab.Column + ") as max from " + csTab.Schema + "." + csTab.Table
		sqlCont := "select min(" + csTab.Column + ") as min, max(" + csTab.Column + ") as max from " + csTab.Schema + "." + csTab.Table
		if colSubQuery != "" {
			sqlCont += " WHERE 1=1 " + colSubQuery
		}
		sqlRange = sqlCont
	}
	LoggerFile.Info("Task: " + csTab.Seq + ", Range SQL: " + sqlRange)
	rangSet := mysql.QueryForSet(srcDB, sqlRange) // 以源端索引情况为准
	if len(rangSet) == 1 {
		if len(rangSet[0]) == 2 {
			return rangSet[0]["min"], rangSet[0]["max"]
		} else {
			LoggerFile.Info("Query for " + csTab.Schema + "." + csTab.Table + "'s min/max, the result set is empty...")
		}
	} else {
		LoggerFile.Info("Query for " + csTab.Schema + "." + csTab.Table + "'s min/max, the result set is empty...")
	}
	return "", ""
}

func sqlRebuild(sqlAs []string, colValues []string, colNames []string, q []string) string {
	var sqlCheck string
	// 构造索引除第一列外的查询条件
	if len(colNames) != len(colValues) && len(colNames) < 1 {
		return ""
	}
	var andCond []string
	var subAddCondSQL string

	if len(colValues) > 0 {
		for i := 1; i < len(colNames); i++ {
			andCond = append(andCond, " "+colNames[i]+" >= "+q[i]+""+colValues[i]+""+q[i]+" ")
		}
		if len(andCond) > 0 {
			subAddCondSQL = " AND " + strings.Join(andCond, " AND ")
		}
		if sqlAs[0] == "split" && colValues[0] != "" {
			if strings.Contains(sqlAs[2], "<=") {
				if strings.Contains(sqlAs[2], "('") {
					sqlCheck = sqlAs[1] + " AND " + fmt.Sprintf("%s%s%s%s%s%s", "(", "'", colValues[0], "'", " < ", sqlAs[2][strings.Index(sqlAs[2], "<=")+2:]) + subAddCondSQL + sqlAs[3]
				} else {
					sqlCheck = sqlAs[1] + " AND " + fmt.Sprintf("%s%s%s%s", "(", colValues[0], " < ", sqlAs[2][strings.Index(sqlAs[2], "<=")+2:]) + subAddCondSQL + sqlAs[3]
				}
			}
		} else if sqlAs[0] == "paging" {
			sqlCheck = sqlAs[1] + " AND " + genCond(colNames, colValues, q) + sqlAs[3]
		}
	} else {
		sqlCheck = sqlAs[1] + " AND " + sqlAs[2] + sqlAs[3]
	}
	return sqlCheck
}

func quotM(cType string) string {
	if strings.Contains(cType, "INT") {
		return ""
	}
	return "'"
}

type CountTuple struct {
	cnt  int  // 校验数据量
	ok   bool // 校验是否通过
	exit bool // 校验后是否退出
	//sql  string
}

type CheckSumTuple struct {
	flag     string   // check sum结果集数量是否与chunk size相等，相等意味着需要继续，少于则代表可以结束了
	checksum []string // 查询到的check sum，包含一个chunk size的check sum
}

type CSTab struct {
	Schema          string
	Table           string
	Seq             string   // 任务序列号，即再任务列表中顺序号
	Index           string   // 索引名，从配置文件中获取
	Column          string   // 索引列或单列名称
	Columns         string   // 表中所有列名，用来替换 select *
	ColumnTypes     []string // 指定列类型，可能是多列，用数组存放，与OrderBy数组中的列对应
	ColumnQuots     []string // 列值是否需要引号，除int外均需要引号
	PalallelS       int      // 表内并行度
	ChunkSize       int      // 每次校验数据块大小，即SQL中 limit 后的数值
	DelaySecond     int      // 容忍延迟，源端和目标端配置数据同步时可配置该参数，单位秒
	ValueColumnK    string   // 配置指定的列
	ValueColumnV    string   // 配置指定的列的类型
	SplitColumn     string   // 配置指定的拆分列
	ValueColumnType string   // 指定列类型
	Interval        int64    // 命令行日志输出频率，单位秒
	OrderBy         []string // SQL语句使用的order by列，索引列可能不止一列，首列用来切分，order by需要全部索引列，类型存放在ColumnTypes中

	SDataServer string // shard节点名称，主从节点均可
	TDataServer string // shard节点名称，主从节点均可
	Database    string // shard节点DB名称
	GridFlag    bool   // 通过grid节点（代理节点）连接数据库
}

func convertArray(arr []interface{}) []string {
	result := []string{}
	for _, v := range arr {
		if str, ok := v.(string); ok {
			result = append(result, str)
		}
	}
	return result
}
